package felix.rule.module.transaction;

import felix.rule.api.*;
import felix.rule.api.util.TNodeUtils;
import felix.rule.data.msg.EngineMsg;
import felix.rule.data.plugin.ComponentType;
import lombok.extern.slf4j.Slf4j;


@Slf4j
@RuleNode(
        type = ComponentType.ACTION,
        name = "synchronization start",
        configClazz = SynchronizationNodeConfiguration.class,
        nodeDescription = "Starts synchronization of message processing based on message originator",
        nodeDetails = "This node should be used together with \"synchronization end\" node. \n This node will put messages into queue based on message originator id. \n" +
                "Subsequent messages will not be processed until the previous message processing is completed or timeout event occurs.\n" +
                "Size of the queue per originator and timeout values are configurable on a system level",
        uiResources = {"static/rulenode/rulenode-core-config.js"},
        configDirective = "tNodeEmptyConfig")
public class SynchronizationBeginNode implements TNode {

    private SynchronizationNodeConfiguration config;

    @Override
    public void init(TContext ctx, TNodeConfiguration configuration) throws TNodeException {
        this.config = TNodeUtils.convert(configuration, SynchronizationNodeConfiguration.class);
        if (this.config.getDuration() == null || this.config.getDuration() < 1) {
            this.config = config.defaultConfiguration();
        }
    }

    @Override
    public void onMsg(TContext ctx, EngineMsg msg) {
        log.trace("Msg enters transaction - [{}][{}]", msg.getId(), msg.getType());
        EngineMsg tbMsg = EngineMsg.transformMsg(msg, msg.getType(), msg.getOriginator(), msg.getMetaData(), msg.getData());

        ctx.getRuleChainTransactionService().beginTransaction(tbMsg, config.getDuration(), startMsg -> {
                    log.trace("Transaction starting...[{}][{}]", startMsg.getId(), startMsg.getType());
                }, endMsg -> {
                    log.trace("Transaction ended successfully...[{}][{}]", endMsg.getId(), endMsg.getType());
                    ctx.tellSuccess(endMsg);
                },
                throwable -> {
                    log.trace("Transaction failed! [{}][{}]", tbMsg.getId(), tbMsg.getType(), throwable);
                    ctx.tellFailure(tbMsg, throwable);
                });
    }

    @Override
    public void destroy() {

    }
}
